package com.k2data.client;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadLocalRandom;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created by luoqifei on 2016/12/28.
 */
public class KafkaClient {
    public static final Logger LOG = LoggerFactory.getLogger(KafkaClient.class);
    private Producer<String, String> producer;
    private String topicName = "";
    private Properties props;

    public KafkaClient(String topicName, String serverAddress) throws ExecutionException, InterruptedException {
        this.topicName = topicName;
        props = new Properties();
        props.put("bootstrap.servers", serverAddress);
        initDefaultProps();
        this.producer = new KafkaProducer<String, String>(props);
    }
    private void testKafka() throws ExecutionException, InterruptedException {
        producer.send(new ProducerRecord<String, String>("huineng-test", "health test", "health test")).get();
    }
    private void initDefaultProps() {
        //props.put("bootstrap.servers", "localhost:9092");
        //Set acknowledgements for producer requests.
        props.put("acks", "1");
        //If the request fails, the producer can automatically retry,
        props.put("retries", 3);
        //props.put("producer.type", "async");
        //wait for all server broker ack in 20s,if not,return timeout,but not contain the net refuse
        props.put("timeout.ms", 20000);
        //Reduce the no of requests less than 0
        //props.put("linger.ms", 1);

        //The buffer.memory controls the total amount of memory available to the producer for buffering.
        //props.put("buffer.memory", 33554432);
        props.put("key.serializer", StringSerializer.class);
        props.put("value.serializer", StringSerializer.class);
    }
    /**
     * @deprecated use sendMessageSync or sendMessageAsync
     */
    public void sendMessage(String key, String values) throws ExecutionException, InterruptedException {
        producer.send(new ProducerRecord<String, String>(topicName, key, values)).get();
    }

    public void sendMessageSync(String key, String values) throws ExecutionException, InterruptedException {
        producer.send(new ProducerRecord<String, String>(topicName, key, values)).get();
    }

    public void sendMessageAsync(String key, String values) throws ExecutionException, InterruptedException {
        producer.send(new ProducerRecord<String, String>(topicName, key, values));
    }

    public void close(){
        if(producer != null){
            producer.close();
        }
    }

    public static void main(String[] args)throws Exception {
      //  KafkaClient client = new KafkaClient("wangyn-test", "127.0.0.1:9092");
      //  KafkaClient client = new KafkaClient(args[0], args[1]);
        long st = DateTime.now().getMillis();
        long et = st;
        long num = 0l;
        /*while ((et-st)<20000){
            et = DateTime.now().getMillis();
            String randomSuffix = String.valueOf(ThreadLocalRandom.current().nextInt(10000, 100000));
            abnormalDataStr = "sediment|2017-03-17 10:55:14.561|(realtimedata|220806010|2017-01-17 10:55:14.514,0.00,0.83,-161.03,-151.73,0.00,89.65,89.65,89.65,6383.00,26.01,25.14,0.00,0.10,0.00,0.10,0.01,0.00,0.01,0.00,-10.69,-9.80,-9.10,-10.60,-10.19,-9.39,-9.15,358.89,359.87,355.23,2.63,1.31,2.63,-2.54,49.98,5.98,0.00,2.50,0.40,-10.50,0.00,-7.19,-10.60,-10.80,-10.00,-0.50,-1.20,0.50,3.79,9.30,4.30,-1.89,-0.50,-1.39,3.79,9.30,4.30,59.64,58.83,59.15,29.66,29.47,29.49,-0.14,0.50,0.53,54.77,54.74,54.72,0.00,0.00,0.00,32627.78,32154.65,270.43,30396.18,24590.83,11036711.000,33078.00,196.67,43.00,27.37,27.27,27.08,27.03,26.83,26.98,26.93,26.88,27.03,27.03,27.03,27.37,-11.30,8.89,-10.30,-10.69,-11.50,-2.79,-4.69,504.16,-502.11,3.29,1.62,138.43,0.00,0.00,0.00,0.00,2.50,600,0,0,0,17388.33,2233.86,0.43,2.02,2231.60,0.00,0.00,14766.87,0,0.00,0.00,0.00,0.00,0.00,0,1500,233,0,0,21122,21122,21122,33,2017,1,17,10,59,51,3,0,0,0,0,0,0,0,0,0,0,0,0,0,170112,709,170111,1936,0,0,0,0,0,0,0,0,0,0,0,3,1,3,TRUE,FALSE,FALSE,FALSE,FALSE,TRUE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,TRUE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,TRUE,FALSE,FALSE,FALSE,TRUE,TRUE,FALSE,FALSE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,TRUE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,TRUE,FALSE,FALSE,TRUE,TRUE,TRUE,FALSE,FALSE,TRUE,TRUE,FALSE,FALSE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,0,0,2017-01-17 10:59:51,2017-01-12 07:09:00,2017-01-11 19:36:00,False,False,False,False,False,False,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,111160128,0,0,0,0,0,0,0,0,976763488,11036711.000,33078,0,0,0,0,0,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,0.00,0.00,0,0,0,161.03,0.82,0.66,-3,0,0,FALSE";
            client.sendMessage("7s_realtime" + randomSuffix, abnormalDataStr);
            num++;
        }*/
        List<Thread> threads = new ArrayList<Thread>();
        for(int i=0; i<10; i++){
            Thread t = new Thread(new Runnable() {
                @Override
                public void run() {
                    KafkaClient client = null;
                    try {
                        client = new KafkaClient("defaultUser-defaultSpace-exception", "192.168.130.83:9092");
                    for(int i=0 ; i<10000; i++){
                        String randomSuffix = String.valueOf(ThreadLocalRandom.current().nextInt(10000, 100000));
                        String abnormalDataStr = "sediment|2017-03-17 10:55:14.561|(realtimedata|2208060"+i+"|2017-01-17 10:55:14.514,0.00,0.83,-161.03,-151.73,0.00,89.65,89.65,89.65,6383.00,26.01,25.14,0.00,0.10,0.00,0.10,0.01,0.00,0.01,0.00,-10.69,-9.80,-9.10,-10.60,-10.19,-9.39,-9.15,358.89,359.87,355.23,2.63,1.31,2.63,-2.54,49.98,5.98,0.00,2.50,0.40,-10.50,0.00,-7.19,-10.60,-10.80,-10.00,-0.50,-1.20,0.50,3.79,9.30,4.30,-1.89,-0.50,-1.39,3.79,9.30,4.30,59.64,58.83,59.15,29.66,29.47,29.49,-0.14,0.50,0.53,54.77,54.74,54.72,0.00,0.00,0.00,32627.78,32154.65,270.43,30396.18,24590.83,11036711.000,33078.00,196.67,43.00,27.37,27.27,27.08,27.03,26.83,26.98,26.93,26.88,27.03,27.03,27.03,27.37,-11.30,8.89,-10.30,-10.69,-11.50,-2.79,-4.69,504.16,-502.11,3.29,1.62,138.43,0.00,0.00,0.00,0.00,2.50,600,0,0,0,17388.33,2233.86,0.43,2.02,2231.60,0.00,0.00,14766.87,0,0.00,0.00,0.00,0.00,0.00,0,1500,233,0,0,21122,21122,21122,33,2017,1,17,10,59,51,3,0,0,0,0,0,0,0,0,0,0,0,0,0,170112,709,170111,1936,0,0,0,0,0,0,0,0,0,0,0,3,1,3,TRUE,FALSE,FALSE,FALSE,FALSE,TRUE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,TRUE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,TRUE,FALSE,FALSE,FALSE,TRUE,TRUE,FALSE,FALSE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,TRUE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,TRUE,FALSE,FALSE,TRUE,TRUE,TRUE,FALSE,FALSE,TRUE,TRUE,FALSE,FALSE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,0,0,2017-01-17 10:59:51,2017-01-12 07:09:00,2017-01-11 19:36:00,False,False,False,False,False,False,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,111160128,0,0,0,0,0,0,0,0,976763488,11036711.000,33078,0,0,0,0,0,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,0.00,0.00,0,0,0,161.03,0.82,0.66,-3,0,0,FALSE";
                        client.sendMessageAsync("7s_realtime" + randomSuffix, abnormalDataStr);
                    }
                    } catch (ExecutionException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
            t.start();
            threads.add(t);
        }
        for (Thread thread : threads) {
            thread.join();
        }
        et = DateTime.now().getMillis();
        LOG.info("time:"+(et-st));
        LOG.info("1 minute send msg num: "+num);
    }

}
